package com.xctech.yace.util;


import com.xctech.yace.enumeration.KafkaRoleType;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class KafkaKerberosUtil {
    public static final String producePrefix = "to";
    public static final String enable = "kafka.sasl.enable";
    public static final String protocol = "security.protocol";
    public static final String mechanism = "sasl.mechanism";
    public static final String jaasConfig ="sasl.jaas.config";
    public static final String saslUser ="sasl.plain.username";
    public static final String saslPasswd ="sasl.plain.password";
    public static final String SASL_JAAS_CONFIG_BEGIN = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"?\" password=\"?\";";

    public static final String PROPERTIES = "properties";
    public static final String SPLIT = ".";
    public static final String FLINKSQL_KAFKASASL_PROPERTIES = "kafka.sasl.conf";




    /**
     * 获取kafka安全认证配置
     * @return
     */
    public static Map<String, String> getKafkaSaslPlainConfMap(Properties properties, KafkaRoleType role) {
        String enableTmp = enable;
        String protocolTmp = protocol;
        String mechanismTmp = mechanism;
        String saslUserTmp = saslUser;
        String saslPasswdTmp = saslPasswd;
        if(KafkaRoleType.PRODUCE.getKafkaRoleName().equals(role.getKafkaRoleName())){
            enableTmp = String.join(SPLIT,producePrefix , enable);
            protocolTmp = String.join(SPLIT,producePrefix , protocol);
            mechanismTmp = String.join(SPLIT,producePrefix , mechanism);
            saslUserTmp = String.join(SPLIT,producePrefix , saslUser);
            saslPasswdTmp = String.join(SPLIT,producePrefix , saslPasswd);
        }
        // 默认不打开安全认证
        String isNeedSecurity = properties.getProperty(enableTmp, "-1");
        // 开启Sasl安全认证
        if (isNeedSecurity.equals("1")) {
            HashMap<String, String> hashMap = new HashMap<>();
            String v1 = properties.getProperty(protocolTmp);
            String v2 = properties.getProperty(mechanismTmp);
            String v3 = properties.getProperty(saslUserTmp);
            String v4 = properties.getProperty(saslPasswdTmp);
            String jasyptPw = properties.getProperty(AuthSaslPlainUtil.jasyptPwConf,"etl");
            String v5 = AuthSaslPlainUtil.replaceEnPassword(jasyptPw, v4);
            String sasljaasconfig = SASL_JAAS_CONFIG_BEGIN.replaceFirst("\\?", v3).replaceFirst("\\?", v5);
            hashMap.put(protocol, v1);
            hashMap.put(mechanism, v2);
            hashMap.put(jaasConfig, sasljaasconfig);

//            // 配置检查
//            hashMap.entrySet().forEach(e->{
//                CheckUtil.matchAndThrowException(e.getValue(), null, "配置开启了sasl-plain认证，但是参数【" + e.getKey() + "】未配置");});

            return hashMap;
        } else return null;

    }

}
